package com.lyrcsoft.webflux_01.reactive;


import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class ReactiveProcessor extends SubmissionPublisher<String> implements Flow.Processor<String, String> {


    // 将关系那出来，后续可能需要用到
    Flow.Subscription subscription;

    @Override
    public void onSubscribe(Flow.Subscription subscription) { // 入参 订阅协议
        System.out.println("processor 订阅关系");
        this.subscription = subscription;
        this.subscription.request(1);
    }

    @Override
    public void onNext(String item) {
        System.out.println("Processor 接收数据：" + item);
        // 中间处理，接收到消息后有自己的处理，这个是Prosessor的不同，我们在学Spring时候的，有很多Processor
        // 在Spring里面叫做后置处理，这里其实也可以这么理解，就是一个中继器，一般情况就是发布者，处理者，
        // 但是在实际业务中，一个流程有很多处理节点，所以需要一个或者多个中继器，来衔接这些业务点，参数要流转下去
        this.submit("Processor 处理：" + item);
        // 每次接收数据 都走onNext，发100接100
        this.subscription.request(1);
        //this.subscription.cancel();
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("Processor onError：" + throwable.getMessage());
        this.subscription.cancel();
    }

    @Override
    public void onComplete() {
        System.out.println("Processor 结束：onComplete");
        this.subscription.cancel();
        // 这个流程完毕之后关闭发布者
        this.close();
        System.out.println("Processor 关闭");
    }
}
